package com.jintian.smart.kernel.switching.api;

import com.jintian.smart.kernel.common.spring.BeanSelfAware;
import com.jintian.smart.kernel.common.switching.FileChangeAdapter;
import com.jintian.smart.kernel.common.switching.FileChangeListener;
import com.jintian.smart.kernel.common.switching.FileChangeMonitor;
import com.jintian.smart.kernel.common.util.StringUtil;
import com.jintian.smart.kernel.module.Model;
import com.jintian.smart.kernel.redis.lock.RedissonLock;
import com.jintian.smart.kernel.switching.db.ExtractRule;
import com.jintian.smart.kernel.switching.dto.DataPacket;
import com.jintian.smart.kernel.switching.dto.FileDataPacket;
import com.jintian.smart.kernel.switching.dto.TransferMessage;
import com.jintian.smart.kernel.switching.dto.db.DBDataPacket;
import com.jintian.smart.kernel.switching.entity.ETLExtractJob;
import com.jintian.smart.kernel.switching.entity.ETLLoadJob;
import com.jintian.smart.kernel.switching.event.AfterSwitchingLoadEvent;
import com.jintian.smart.kernel.switching.mapper.IETLLoadJobMapper;
import lombok.extern.slf4j.Slf4j;
import net.lingala.zip4j.ZipFile;
import org.beetl.sql.core.mapping.StreamData;
import org.beetl.sql.core.query.LambdaQuery;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.File;
import java.io.IOException;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.*;

@Slf4j
public abstract class AbstractLoaderTask implements ILoadTask, BeanSelfAware, ApplicationContextAware {

    @Value("${mda.switching.load.read-path}")
    private String readPath;

    @Value("${mda.switching.load.response-path}")
    private String responsePath;

    @Value("${mda.switching.load.log-path:#{null}}")
    private String logPath;

    @Value("${mda.switching.load.retries:60}")
    private int retries = 60;

    @Value("${mda.switching.lock:false}")
    private boolean lock;

    @Value("${mda.switching.extract.sequence:true}")
    protected boolean sequence = true;

    private FileChangeListener transferMessageFileListener;

    @Autowired
    private IETLLoadJobMapper loadJobMapper;

    private ApplicationContext applicationContext;

    @Autowired
    private RedissonLock redissonLock;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;

    }

    private AbstractLoaderTask self;

    @Override
    public void setSelf(Object proxyBean) {
        this.self = (AbstractLoaderTask) proxyBean;
    }

    protected File getDataTypeReadDir() {
        return new File(readPath, this.getDataType().getName());
    }

    protected File getResponseWritePath() {
        return new File(responsePath, this.getDataType().getName());
    }

    protected File getLogPath() {
        if (StringUtil.isBlank(this.logPath)) {
            return null;
        }
        return new File(this.logPath, this.getDataType().getName());

    }

    @EventListener
    protected void onApplicationRefreshEvent(ContextRefreshedEvent event) {
        File dataTypeReadDir = this.getDataTypeReadDir();
        if (event.getApplicationContext() == this.applicationContext) {
            try {
                Files.walkFileTree(dataTypeReadDir.toPath(), new SimpleFileVisitor<Path>() {
                    @Override
                    public FileVisitResult visitFile(Path path, BasicFileAttributes attrs) throws IOException {
                        self.createETLLoadJob(path);
                        return FileVisitResult.CONTINUE;
                    }
                });
            } catch (IOException ex) {
                log.error("{}初始化失败", this.getDataType(), ex);
            }
        }
    }

    @PostConstruct
    private void init() {
        final FileChangeAdapter transferMessageFileListener = new FileChangeAdapter() {
            @Override
            public void fileCreated(Path file) {
                if (Files.isDirectory(file)) {
                    // 级联监听
                    watchDir(file, this);
                    try {
                        Files.find(file, Integer.MAX_VALUE, (path, attr) -> {
                            return !attr.isDirectory() && path.toFile().getName().toLowerCase()
                                    .endsWith(IExtractTask.TRANSFER_DMP_FILE_EXT);
                        }).forEach((f) -> self.createETLLoadJob(f));
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                } else {
                    self.createETLLoadJob(file);
                }
            }

            @Override
            public void fileModified(Path file) {
                // System.out.println("modified " + file);
                // self.createETLLoadJob(file);
            }

            @Override
            public void fileDeleted(Path file) {
                if (Files.isDirectory(file)) {
                    FileChangeMonitor.unregister(this, file);
                }
            }
        };
        this.transferMessageFileListener = transferMessageFileListener;
        File dataTypeReadDir = this.getDataTypeReadDir();
        if (!dataTypeReadDir.exists()) {
            dataTypeReadDir.mkdirs();
        }
        this.watchDir(dataTypeReadDir.toPath(), transferMessageFileListener);

    }

    private void watchDir(Path dir, FileChangeAdapter dataPacketFileListener) {
        // 递归监听所有子
        try {
            Files.walkFileTree(dir, new SimpleFileVisitor<Path>() {
                @Override
                public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
                    FileChangeMonitor.register(dataPacketFileListener, dir);
                    log.info("watch dir:{}", dir);
                    return super.postVisitDirectory(dir, exc);
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @PreDestroy
    private void destroy() {
        if (this.transferMessageFileListener != null) {
            FileChangeMonitor.unregister(this.transferMessageFileListener);
        }
    }

    private String getInputResource(Path file) {
        Path exchangeDir = this.getDataTypeReadDir().toPath();
        // 规则
        String path = file.subpath(exchangeDir.getNameCount(), file.getNameCount()).toString();
        return path;
    }

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void createETLLoadJob(Path fullPath) {
        if (Files.isDirectory(fullPath)) {
            return;
        }
        if (fullPath.toFile().getName().toLowerCase().endsWith(IExtractTask.TRANSFER_DMP_FILE_EXT)) {
            this.createETLLoadJobByDmpFile(fullPath.toFile());
        } else if (fullPath.toFile().getName().toLowerCase().endsWith(IExtractTask.TRANSFER_DZP_FILE_EXT)) {
            this.createETLLoadJobByDzpFile(fullPath.toFile());
        }
    }

    private void createETLLoadJobByDmpFile(File dmpFile) {
        try (TransferMessage msg = TransferMessage.loadFromFile(dmpFile)) {
            switch (msg.getHeader().getMessageType()) {
                case DataPacket:
                case FileDataPacket:
                    DataPacket info = TransferMessage.readMessageBody(msg.getMessageBodyBuf(), DataPacket.class);
                    this.createETLLoadJob(info, dmpFile);
                    break;
                default:
                    log.error("createETLLoadJob {} error", msg.getHeader().getMessageType());
            }
        } catch (Exception e) {
            log.error("read TransferMessage error", e);
        }
    }

    private void createETLLoadJobByDzpFile(File dzpFile) {
        ZipFile zip = new ZipFile(dzpFile);
        String extractDir = dzpFile.getParent();
        try {
            zip.extractAll(extractDir);
            zip.close();
            dzpFile.delete();
        } catch (Exception e) {
            if (e instanceof RuntimeException) {
                throw (RuntimeException) e;
            }
            throw new IllegalArgumentException(e);
        }
    }

    private void createETLLoadJob(DataPacket info, File dmpFile) {
        String ruleName = info.getRuleName();
        String dataType = info.getDataType().substring(info.getDataType().lastIndexOf(".") + 1);
        String inputResource = this.getInputResource(dmpFile.toPath());
        ETLLoadJob job = loadJobMapper.createLambdaQuery().andEq(ETLLoadJob::getDataType, dataType)
                .andEq(ETLLoadJob::getRuleName, ruleName).andEq(ETLLoadJob::getInputResource, inputResource).single();
        if (job == null) {
            job = new ETLLoadJob();
            job.setCreateTime(LocalDateTime.now());
            job.setRetries(0);
            job.setRuleName(ruleName);
            job.setInputResource(inputResource);
            job.setDataType(dataType);
        }
        job.setStatus(ETLLoadJob.Status.waiting.name());
        loadJobMapper.upsert(job);
        if (log.isDebugEnabled())
            log.debug("create Load Job {}", dmpFile);
    }

    @Override
    @Transactional
    public void execute() {
        // dataType = this.dataType order by ruleName,createTime
        LambdaQuery<ETLLoadJob> query = loadJobMapper.createLambdaQuery()
                .andEq(ETLLoadJob::getDataType, this.getDataType().getSimpleName()).order(ETLLoadJob::getRuleName, true)
                .order(ETLLoadJob::getCreateTime, true);
        if (System.getProperty("switching-ruleNames") != null) {
            // and ruleName in ()
            query.andIn(ETLLoadJob::getRuleName, Arrays.asList(System.getProperty("switching-ruleNames").split(",")));
        }
        // and (status = 'waiting' or status='error' and retries<60)
        query.and(query.condition().andEq(ETLLoadJob::getStatus, ETLLoadJob.Status.waiting.name())
                .or(query.condition().andEq(ETLLoadJob::getStatus, ETLLoadJob.Status.error.name())
                        .andLess(ETLLoadJob::getRetries, this.retries)));
        StreamData<ETLLoadJob> stream = query.stream();

        Set<String> nolockIgnoreRules = new HashSet<>();
        Set<String> sequenceErrorIgnoreRules = new HashSet<>();
        List<String> lockRules = new ArrayList<>();
        try {
            ETLLoadJob job;
            while ((job = stream.next()) != null) {
                if (this.sequence && sequenceErrorIgnoreRules.contains(job.getRuleName())) {
                    // 顺序执行、忽略ignoreSequenceRules
                    continue;
                }
                if (nolockIgnoreRules.contains(job.getRuleName())) {
                    // 无规则锁忽略
                    continue;
                }
                if (redissonLock != null && lock) {
                    if (redissonLock.lock(this.getClass().getName() + ":" + job.getRuleName(), 60)) {
                        // 记录锁的规则名，最后需释放
                        lockRules.add(job.getRuleName());
                    } else {
                        // 添加无锁忽略
                        nolockIgnoreRules.add(job.getRuleName());
                        continue;
                    }
                }
                try {
                    this.self.execute(job);
                } catch (Exception ex) {

                }

                if (job.getStatus().equals(ETLLoadJob.Status.error.name())) {
                    this.self.updateJob(job);
                    if (this.sequence) {
                        sequenceErrorIgnoreRules.add(job.getRuleName());
                    }
                } else if (job.getStatus().equals(ETLLoadJob.Status.succeed.name())) {
                    if (this.writeResponse(job)) {
                        // 写响应成功，设置为finish
                        this.self.finishJob(job);
                    } else {
                        this.self.updateJob(job);
                    }
                } else {
                    this.self.updateJob(job);
                }
            }
        } finally {
            for (String ruleName : lockRules) {
                redissonLock.release(this.getClass().getName() + ":" + ruleName);
            }
        }

    }

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void updateJob(ETLLoadJob job) {
        this.loadJobMapper.updateById(job);
    }

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void finishJob(ETLLoadJob job) {
        job.setStatus(ETLLoadJob.Status.finish.name());
        this.loadJobMapper.createLambdaQuery().set(ETLLoadJob::getStatus, ETLLoadJob.Status.finish.name())
                .andEq(ETLLoadJob::getId, job.getId()).update();
    }

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public boolean execute(ETLLoadJob job) {
        try {
            // 创建加载器
            File readPath = this.getDataTypeReadDir();
            File inputResource = new File(readPath, job.getInputResource());
            DataPacket dp = this.getDataPacket(inputResource);
            File dataFile = null;
            Collection<Model> modelList = this.applicationContext.getBeansOfType(Model.class).values();
            Collection<ILoader> ILoaders = new HashSet<ILoader>();

            for (Model m : modelList) {
                ILoaders.addAll(m.getModelContext().getBeansOfType(ILoader.class).values());
            }
            boolean b = false;
            for (ILoader ILoader : ILoaders) {
                Class ruleClass = Class.forName(dp.getRuleClass(), true, applicationContext.getClassLoader());
                if (ILoader.support(this.getDataType(), ruleClass, dp.getRuleName())) {
                    b = true;
                    job.setExecuteTime(LocalDateTime.now());

                    Object loadData = null;
                    if (dp instanceof FileDataPacket) {
                        dataFile = new File(inputResource.getParent(), ((FileDataPacket) dp).getFileName());
                        // TODO 获取解析数据，给事件传值，否则事件里获取不到数据
                        loadData = ILoader.load(dataFile);
                    } else {
                        loadData = ILoader.load(dp.getData());
                    }

                    AfterSwitchingLoadEvent event = new AfterSwitchingLoadEvent(job, loadData);
                    ILoader.getApplicationContext().publishEvent(event);
                    // 移动到日志
                    if (this.getLogPath() != null) {
                        File logDir = new File(this.getLogPath(),
                                new SimpleDateFormat("yyyy/MM/dd/").format(System.currentTimeMillis()));
                        if (!logDir.exists()) {
                            logDir.mkdirs();
                        }
                        File logFile = new File(logDir, inputResource.getName());
                        if (logFile.exists()) {
                            logFile.delete();
                        }
                        inputResource.renameTo(logFile);

                        if (dataFile != null) {
                            File logDataFile = new File(logDir, dataFile.getName());
                            if (logDataFile.exists()) {
                                logDataFile.delete();
                            }
                            dataFile.renameTo(logDataFile);
                        }
                    } else {
                        inputResource.delete();
                        if (dataFile != null) {
                            dataFile.delete();
                        }
                    }
                    job.setException(null);
                    job.setStatus(ETLLoadJob.Status.succeed.name());
                } else {
                    log.warn("加载器不匹配:{},{},{}", DBDataPacket.class.equals(this.getDataType()), ExtractRule.class.equals(ruleClass),
                            "classpath:etl/loader.rule-path/" + dp.getRuleName() + ".xml");
                }
            }
            if (!b) {
                job.setStatus(ETLExtractJob.Status.error.name());
                job.setRetries(job.getRetries() + 1);
                job.setException("无匹配的加载器,装载器数量:" + ILoaders.size());
            }
            return true;
        } catch (Exception ex) {
            // 3.载入异常
            log.error("{}载入[id={}]失败", this.getDataType(), job.getId(), ex);
            job.setStatus(ETLLoadJob.Status.error.name());
            job.setRetries(job.getRetries() + 1);
            job.setException(ex.getMessage());
            return false;
        }
    }

    private DataPacket getDataPacket(File file) throws IOException {
        try (TransferMessage msg = TransferMessage.loadFromFile(file);) {
            switch (msg.getHeader().getMessageType()) {
                case DataPacket:
                    return TransferMessage.readMessageBody(msg.getMessageBodyBuf(), DataPacket.class);
                case FileDataPacket:
                    return TransferMessage.readMessageBody(msg.getMessageBodyBuf(), FileDataPacket.class);
                default:
                    throw new IllegalArgumentException("暂不支持" + msg.getHeader().getMessageType() + "的消息");
            }
        }
    }

    public static void main(String[] args) throws IOException {
        Files.walkFileTree(new File("D:\\temp").toPath(), new SimpleFileVisitor<Path>() {
            @Override
            public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
                log.info("watch exists dir:{}", dir);
                Files.find(dir, Integer.MAX_VALUE, (path, attr) -> {
                    return !attr.isDirectory();
                }).forEach((f) -> System.out.println(f));
                return super.postVisitDirectory(dir, exc);
            }
        });
    }
}
